package event;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


import java.util.*;

/**
 * kafka发送消息工具类
 */
public class KafkaProducerUtil {
    private static Logger logger = LoggerFactory.getLogger(KafkaProducerUtil.class);
    private static KafkaProducerUtil instance;
    private Producer<Integer, String> producer;
    public static boolean isEnable = true;

    static {
        String isEnableStr = "true";
        if ("false".equalsIgnoreCase(isEnableStr)) {
            isEnable = false;
        }
        logger.info("kafkaEnable->{}", isEnable);
    }

    public KafkaProducerUtil() {
        Properties props = new Properties();
        String broker = "10.0.6.200:9092,10.0.6.200:9093,10.0.6.200:9094";
        props.put("bootstrap.servers", broker);
        props.put("retries", 0);
        props.put("producer.type", "async");
        props.put("max.block.ms", 5000);
        props.put("linger.ms", 1);
        props.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        producer = new KafkaProducer(props);
        logger.info("kafkaProducerCreate->bootstrap.servers{}", broker);
    }

    private static class ResourceHolder {
        public static KafkaProducerUtil resource = new KafkaProducerUtil(); // This will be lazily initialised
    }

    public static KafkaProducerUtil getInstance() {
        return ResourceHolder.resource;
    }

    public static KafkaProducerUtil getInstance2() {
        return ResourceHolder.resource;
    }


    private class DataCallback implements Callback {
        private String topic;
        private Object data;

        public DataCallback(String topic, Object data) {
            this.topic = topic;
            this.data = data;
        }

        public void onCompletion(RecordMetadata metadata, Exception exception) {
            if (null != metadata) {
                logger.debug("sendKafkaSuccess->topic:{},offset:{},data:{}", topic, metadata.offset(), data);
            } else {
                logger.error("sendKafkaError->topic:{},data:{}", topic, data, exception);
            }
        }
    }

    public void sendAsync(String topic, Integer key, Object message) {
        if (isEnable) {
            DataCallback callback = new DataCallback(topic, message);
            producer.send(new ProducerRecord(topic, key, message), callback);
//            Sender sender = new Sender(topic, message);
//            Thread thread = new Thread(sender);
//            thread.start();


        } else {
            logger.info("kafkaNotEnable->topic:{},message:{}", topic, message);
        }
    }

    /**
     * 消费消息
     *
     * @param topic
     */
    public void getMsg(String topic) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "10.0.6.200:9092,10.0.6.200:9093,10.0.6.200:9094");
        props.put("group.id", "test12");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("partitioner.class", "event.MyPartition");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        //consumer.subscribe(Arrays.asList(topic));
        TopicPartition topicPartition = new TopicPartition("TOPIC_SDK_CN_BIND_PHONE", 0);

        consumer.assign(Arrays.asList(topicPartition));
        consumer.seek(topicPartition, 2029);
//        Map<String,String> map = KafkaOffsetSearch.getPartitonAndOffset();
//        System.out.println(map.keySet());
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);

            for (ConsumerRecord<String, String> record : records) {
                // AdEvent adEvent = JSON.parseObject(record.value(), AdEvent.class);
                System.out.println(record.toString());
            }

        }
    }

    private class Sender implements Runnable {
        private String topic;// topic
        private Object message;// 自定义消息

        // construct
        public Sender(String topic, Object message) {
            this.topic = topic;
            this.message = message;
        }

        @Override
        public void run() {
            send(topic, message);
        }
    }

    /**
     * 发送是同步的，如果kafka出问题最大导致发送线程卡5秒
     *
     * @param topic
     * @param message
     */
    public void send(String topic, final Object message) {
        DataCallback callback = new DataCallback(topic, message);

        if (isEnable) {
            producer.send(new ProducerRecord(topic, message), callback);
        } else {
            logger.info("kafkaNotEnable->topic:{},message:{}", topic, message);
        }
    }

    public void close() {
        producer.close();
    }


    /**
     * 消费消息
     *
     * @param topic
     */
    public Set<String> getMsg(String server, String groupId, String topic, long start, long end, Set target, int[] eventType) {
        Properties props = new Properties();
        props.put("bootstrap.servers", server);
        props.put("group.id", groupId);
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList(topic));

        List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);
        System.out.println("partitionInfos:" + partitionInfos);

        Map<TopicPartition, Long> timestampsToSearch = new HashMap();

        for (PartitionInfo partitionInfo : partitionInfos) {
            TopicPartition partition = new TopicPartition(partitionInfo.topic(), partitionInfo.partition());
            timestampsToSearch.put(partition, start);
        }

        System.out.println("开始查找offset:" + timestampsToSearch);
        Map<TopicPartition, OffsetAndTimestamp> offset = consumer.offsetsForTimes(timestampsToSearch);
        if (offset.isEmpty()) {
            System.out.println("指定时间内的找数据未找到");
            return new HashSet();
        } else {
            System.out.println("找到数据offset:" + offset);
        }
        ConsumerRecords<String, String> records = consumer.poll(100);
        for (Map.Entry<TopicPartition, OffsetAndTimestamp> topicPartitionOffsetAndTimestampEntry : offset.entrySet()) {
            if (null == topicPartitionOffsetAndTimestampEntry.getValue()) {//为空说明没有数据
                return new HashSet();
            }
            consumer.seek(topicPartitionOffsetAndTimestampEntry.getKey(), topicPartitionOffsetAndTimestampEntry.getValue().offset());
        }

        Set<Integer> readSuccess = new HashSet();
        Set<String> match = new HashSet();
        Set<String> matchData = new HashSet();
        try {
            long count = 0;
            long beginTime = System.currentTimeMillis();
//            ConsumerRecords<String, String> records;
            while (true) {
                records = consumer.poll(2000);
                for (ConsumerRecord<String, String> record : records) {
                    count++;
                    if (count % 100000 == 0) {
                        System.out.println("读取数据条数：" + count);
                    }
                    if (record == null || record.timestamp() > end) {
                        readSuccess.add(record.partition());
                        continue;
                    }
                    AdEvent adEvent = JSON.parseObject(record.value(), AdEvent.class);
                    if (null != eventType) {
                        boolean isTypeMatch = false;//默认不过滤
                        for (int s : eventType) {
                            if (adEvent.type.type == s) {
                                isTypeMatch = true;
                                break;
                            }
                        }
                        if (!isTypeMatch) {
                            continue;
                        }
                    }

                    String androidId = adEvent.getAndroidId();
                    String idfa = adEvent.getIdfa();
                    String imei = adEvent.getImei();
                    boolean isAdd = false;
                    if (target.contains(androidId)) {
                        isAdd = match.add(androidId);
                    } else if (target.contains(idfa)) {
                        isAdd = match.add(idfa);
                    } else if (target.contains(imei)) {
                        isAdd = match.add(imei);
                    }

                    if (isAdd) {
                        matchData.add(record.value());
                    }
                }
                if (records.isEmpty() || readSuccess.size() >= partitionInfos.size()) {//全部都读完了
                    System.out.println("records isEmpty：" + records.isEmpty() + " readSuccessPartitionSize:" + readSuccess.size());
                    break;
                }
            }
            System.out.println("读取数据总条数：" + count + " useTime:" + (System.currentTimeMillis() - beginTime));
        } finally {
            consumer.close();
        }
        return matchData;
    }
}
